"""Tests for queues.py"""

import unittest
from queue import Empty, Full  # exceptions from standard lib's queue

from tulip import events
from tulip import futures
from tulip import locks
from tulip import queues
from tulip.tasks import coroutine, task


# TODO: looks like Tulip needs a more convenient means of pausing
@coroutine
def pause(delay):
    fut = futures.Future()
    events.get_event_loop().call_later(delay, fut.set_result, None)
    yield from fut


class _QueueTestBase(unittest.TestCase):

    def setUp(self):
        self.event_loop = events.new_event_loop()
        events.set_event_loop(self.event_loop)

    def tearDown(self):
        self.event_loop.close()


class QueueBasicTests(_QueueTestBase):
    def test_repr(self):
        q = queues.Queue()
        self.assertTrue(repr(q).startswith('<Queue'))
        self.assertTrue(hex(id(q)) in repr(q))

        q.get()  # returns a pending future
        self.assertTrue('_getters[1]' in repr(q))

        q = queues.Queue(maxsize=0)
        q.put(1)  # returns a pending future
        self.assertTrue('_putters[1]' in repr(q))

        q = queues.Queue()
        q.put(1)
        self.assertTrue('_queue=[1]' in repr(q))

    def test_str(self):
        q = queues.Queue()
        self.assertTrue(str(q).startswith('<Queue'))

        q.get()  # returns a pending future
        self.assertTrue('_getters[1]' in str(q))

        q = queues.Queue(maxsize=0)
        q.put(1)  # returns a pending future
        self.assertTrue('_putters[1]' in str(q))

        q = queues.Queue()
        q.put(1)
        self.assertTrue('_queue=[1]' in str(q))

    def test_empty(self):
        q = queues.Queue()
        self.assertTrue(q.empty())
        q.put(1)
        self.assertFalse(q.empty())
        self.assertEqual(1, q.get(block=False))
        self.assertTrue(q.empty())

    def test_full(self):
        q = queues.Queue()
        self.assertFalse(q.full())

        q = queues.Queue(maxsize=0)
        self.assertTrue(q.full())

    def test_order(self):
        q = queues.Queue()
        for i in [1, 3, 2]:
            q.put(i)

        items = [q.get(block=False) for _ in range(3)]
        self.assertEqual([1, 3, 2], items)

    def test_max_size(self):
        q = queues.Queue(maxsize=2)
        have_been_put = []

        @task
        def putter():
            for i in range(3):
                yield from q.put(i)
                have_been_put.append(i)

        @task
        def test():
            putter()
            yield from pause(0.1)

            # putter is blocked after putting two items
            self.assertEqual([0, 1], have_been_put)
            self.assertEqual(0, q.get(block=False))

            # Let putter resume and put last item
            yield from pause(0.1)
            self.assertEqual([0, 1, 2], have_been_put)
            self.assertEqual(1, q.get(block=False))
            self.assertEqual(2, q.get(block=False))

        self.event_loop.run_until_complete(test())

    def test_increase_maxsize(self):
        q = queues.Queue(1)

        history = []

        @task
        def putter():
            yield from q.put(0)  # doesn't block
            history.append('put 0')
            yield from q.put(1)  # blocks
            history.append('put 1')

            # doesn't block; runs after maxsize is increased to 3
            yield from q.put(2)
            history.append('put 2')

            try:
                yield from q.put(3, timeout=0.1)
            except Full:
                pass
            else:
                self.fail("put() didn't time out")

            history.append('failed to put 3')
            yield from q.put(4)  # blocks
            history.append('put 4')

        @task
        def test():
            # start task
            putter_task = putter()
            yield from pause(0.2)
            # now putter is blocked at 'put 1'
            history.append('setting maxsize 3')
            # unblock it, it'll block at 'put 3'
            q.maxsize = 3
            # let 'put 3' task time out
            yield from pause(0.2)
            history.append('set maxsize 3')

            self.assertEqual(3, q.qsize())  # 3 wasn't put, timed out

            history.append('setting maxsize 4')
            # unblock it
            q.maxsize = 4
            yield from putter_task

        self.event_loop.run_until_complete(test())

        self.assertEqual([
            'put 0',
            'setting maxsize 3', 'put 1', 'put 2',
            'failed to put 3', 'set maxsize 3',
            'setting maxsize 4', 'put 4'
        ], history)

    def test_decrease_maxsize(self):
        q = queues.Queue(1)
        q.put(0)

        def decrease():
            q.maxsize -= 1

        self.assertRaises(RuntimeError, decrease)
        q.maxsize = 2
        q.maxsize = 1  # no error
        q.maxsize = 2
        q.put(1)
        self.assertRaises(RuntimeError, decrease)


class QueueGetTests(_QueueTestBase):

    def test_blocking_get_nowait(self):
        q = queues.Queue()
        q.put(1)

        @task
        def queue_get():
            return (yield from q.get())

        res = self.event_loop.run_until_complete(queue_get())
        self.assertEqual(1, res)

    def test_blocking_get_wait(self):
        q = queues.Queue()
        started = locks.EventWaiter()
        finished = False

        @task
        def queue_get():
            nonlocal finished
            started.set()
            res = yield from q.get()
            finished = True
            return res

        @task
        def queue_put():
            self.event_loop.call_later(.1, q.put, 1)
            queue_get_task = queue_get()
            yield from started.wait()
            self.assertFalse(finished)
            res = yield from queue_get_task
            self.assertTrue(finished)
            return res

        res = self.event_loop.run_until_complete(queue_put())
        self.assertEqual(1, res)

    def test_nonblocking_get(self):
        q = queues.Queue()
        q.put(1)
        self.assertEqual(1, q.get(block=False))

    def test_nonblocking_get_exception(self):
        q = queues.Queue()
        self.assertRaises(Empty, q.get, block=False)

    def test_get_timeout(self):
        q = queues.Queue()

        @task
        def queue_get():
            return (yield from q.get(timeout=.1))

        self.assertRaises(
            Empty,
            self.event_loop.run_until_complete,
            queue_get())


class QueuePutTests(_QueueTestBase):

    def test_blocking_put_nowait(self):
        q = queues.Queue()

        @task
        def queue_put():
            # No maxsize, won't block
            return (yield from q.put(1))

        res = self.event_loop.run_until_complete(queue_put())
        self.assertTrue(res)

    def test_blocking_put_wait(self):
        q = queues.Queue(maxsize=0)
        started = locks.EventWaiter()
        finished = False

        @task
        def queue_put():
            nonlocal finished
            started.set()
            res = yield from q.put(1)
            finished = True
            return res

        @task
        def queue_get():
            self.event_loop.call_later(.1, q.get)
            queue_put_task = queue_put()
            yield from started.wait()
            self.assertFalse(finished)
            res = yield from queue_put_task
            self.assertTrue(finished)
            return res

        res = self.event_loop.run_until_complete(queue_get())
        self.assertEqual(True, res)

    def test_nonblocking_put(self):
        q = queues.Queue()
        self.assertTrue(q.put(1, block=False))
        self.assertEqual(1, q.get(block=False))

    def test_nonblocking_put_exception(self):
        q = queues.Queue(maxsize=0)
        self.assertRaises(Full, q.put, 1, block=False)

    def test_put_timeout(self):
        q = queues.Queue(maxsize=0)

        @task
        def queue_put():
            return (yield from q.put(1, timeout=.1))

        self.assertRaises(
            Full,
            self.event_loop.run_until_complete,
            queue_put())


class LifoQueueTests(_QueueTestBase):

    def test_order(self):
        q = queues.LifoQueue()
        for i in [1, 3, 2]:
            q.put(i)

        items = [q.get(block=False) for _ in range(3)]
        self.assertEqual([2, 3, 1], items)


class PriorityQueueTests(_QueueTestBase):

    def test_order(self):
        q = queues.PriorityQueue()
        for i in [1, 3, 2]:
            q.put(i)

        items = [q.get(block=False) for _ in range(3)]
        self.assertEqual([1, 2, 3], items)


class JoinableQueueTests(_QueueTestBase):

    def test_task_done_underflow(self):
        q = queues.JoinableQueue()
        self.assertRaises(q.task_done)

    def test_task_done(self):
        q = queues.JoinableQueue(initial=range(100))
        accumulator = 0

        # Two workers get items from the queue and call task_done after each.
        # Join the queue and assert all items have been processed.

        @task
        def worker():
            nonlocal accumulator

            while True:
                item = yield from q.get()
                accumulator += item
                q.task_done()

        @task
        def test():
            for _ in range(2):
                worker()

            yield from q.join()

        self.event_loop.run_until_complete(test())
        self.assertEqual(sum(range(100)), accumulator)

    def test_join_empty_queue(self):
        q = queues.JoinableQueue()

        # Test that a queue join()s successfully, and before anything else
        # (done twice for insurance).

        @task
        def join():
            yield from q.join()
            yield from q.join()

        self.event_loop.run_until_complete(join())

    def test_join_timeout(self):
        q = queues.JoinableQueue()
        q.put(1)

        @task
        def join():
            yield from q.join(.1)

        # Join completes in ~ .1 seconds, although no one calls task_done()
        self.event_loop.run_until_complete(join())


if __name__ == '__main__':
    unittest.main()
